package com.patsnap.data.npd.dw.etl.job.ods.parser.journal;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.patsnap.data.npd.dw.etl.job.ods.dispatcher.DataLayer;
import com.patsnap.data.npd.dw.etl.job.ods.dispatcher.ProcessorIdentifier;
import com.patsnap.data.npd.dw.etl.job.ods.serialization.PreOdsInfo;
import com.patsnap.data.npd.dw.repository.DBOperator;
import com.patsnap.data.npd.dw.schema.ods.journal.OdsJournalJBestInfo;

import java.util.Map;

@ProcessorIdentifier(parser = "JOURNAL_PARSER_BEST", type = "npd_journal", layer = DataLayer.ODS)
public class OdsJBestJournalDataProcessor extends OdsJournalDataProcessor<PreOdsInfo> {

    public OdsJBestJournalDataProcessor(Map<String, Object> globalConfig, DBOperator dbOperator) {
        super(globalConfig, dbOperator);
    }

    @Override
    public PreOdsInfo processHandle(String op, PreOdsInfo in) throws Exception {
        return null;
    }

    @Override
    public String layerDependRelation() {
        ProcessorIdentifier annotation = this.getClass().getAnnotation(ProcessorIdentifier.class);
        return annotation.layer().name() + "_" + annotation.type() + "_" + annotation.parser() + " To Ods";
    }

    @Override
    public void launchWithJsonRecord(PreOdsInfo preOdsInfo) {
        try {
            String body = preOdsInfo.getBody();
            JSONObject jo = JSON.parseObject(body);
            OdsJournalJBestInfo odsJournalJBestInfo = jo.toJavaObject(OdsJournalJBestInfo.class);
            operator.getOdsJournalRepo().upsert(odsJournalJBestInfo, OdsJournalJBestInfo.class);
        } catch (Exception e) {
            throw new RuntimeException("Error launch, info => " + JSON.toJSONString(preOdsInfo), e);
        }
    }
}
